package com.kafka.producer;

import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import net.sf.json.JSONObject;

import com.kafka.bean.MedicalInfo;
import com.kafka.common.Constant;
import com.kafka.common.String.StringValues;
import com.kafka.core.BaseProducer;
import com.kafka.dao.MedicalInfoMapper;
import com.kafka.enumtype.MsgType;
import com.kafka.producer.inter.IProducer;
import com.kafka.util.BaseUtil;
import com.kafka.valid.MedicalValid;

/**
 * 注册消息类
 * 
 * @author leihongan
 * 
 */
public class MedicalProducer extends BaseProducer implements IProducer {
	MedicalInfoMapper medicalDao = sqlSession.getMapper(MedicalInfoMapper.class);
	private final Producer<String, String> producer;
	public String TOPIC;

	public MedicalProducer() {
		Properties properties = new Properties();
		try {
			properties.load(ClassLoader
					.getSystemResourceAsStream("medical.properties"));
			TOPIC = properties.getProperty("topic");
		} catch (IOException e) {
			e.printStackTrace();
		}
		producer = new Producer<String, String>(new ProducerConfig(properties));
	}

	public String produce(JSONObject message) {
		if (MsgType.MEDICAL.toString().equals(message.getString("msgtype"))){
			return MedicalProduce(message);
		} else {
			return null;
		}
	}
	
	public String MedicalProduce(JSONObject message) {
		try {
			String msgId = message.getString("msgid");
			// 校验数据合法性
			JSONObject ck = MedicalValid.validUpload(rebuildJSON(message.getJSONObject("datas")), msgId);
			if (ck != null) {
					addMessage(message,TOPIC,Constant.FAIL, ck.getString("msg"));
					// 如果错误信息不为空，返回错误信息
					return ck.toString();
			} else {
				// 追加appid和secretkey
				addAppidAndSecretkey(message);
				Map<String, Object> params = new HashMap<>();
				params.put("status", Constant.SUCCESS);
				params.put("message_id", message.get("msgid"));
				if(messageDao.selectByParam(params)==0){
					JSONObject tmp = new JSONObject();
					tmp = JSONObject.fromObject(message.toString());
					// 发送消息keyedMessage第一个参数是topic第二个参数就是发送的消息内容
					producer.send(new KeyedMessage<String, String>(TOPIC, rebuildSendMessage(message).toString()));
					//先发送再存储，如果发送异常则不会执行存储代码
					addMedicalInfo(tmp);
					log.info("###MedicalProducer:" + message.toString() + ",time:" + BaseUtil.getDate());
					addMessage(message,TOPIC,Constant.SUCCESS, StringValues.SEND_SUCCESSFUL);
					return successMsg(message.get("msgid"));
				}else{
					return failMsg(message.get("msgid"),StringValues.SEND_FAIL);
				}
			}
		} catch (Exception e) {
			addMessage(message,TOPIC,Constant.FAIL, e.toString());
			return failMsg(message.get("msgid"), e.toString());
		}
	}

	private JSONObject rebuildSendMessage(JSONObject message) {
		JSONObject datas = message.getJSONObject("datas");
		JSONObject result = new JSONObject();
		result.put("uuid", datas.get("uuid"));
		result.put("type", datas.getInt("type"));
		result.put("value", datas.get("value"));
		result.put("datetime", datas.get("datetime"));
		result.put("shopcode", datas.get("shopcode"));
		result.put("shopname", datas.get("shopname"));
		message.put("datas", result);
		return message;
	}
	
//	private int getMedicalType(int type){
//		switch (type) {
//		case 0:
//			return Integer.valueOf(MedicalType.GLU.toString());
//		case 1:
//			return Integer.valueOf(MedicalType.UA.toString());
//		case 2:
//			return Integer.valueOf(MedicalType.CHOL.toString());
//		default:
//			return 0;
//		}
//	}

	private void addMedicalInfo(JSONObject message) throws ParseException {
		JSONObject medicalInfoData = message.getJSONObject("datas");
		MedicalInfo medicalInfo = createEntity(medicalInfoData);
		medicalDao.insert(medicalInfo);
		sqlSession.commit();
	}

	private MedicalInfo createEntity(JSONObject medicalInfoData) throws ParseException {
		MedicalInfo medicalInfo = new MedicalInfo();
		medicalInfo.setUuid(medicalInfoData.getString("uuid"));
		medicalInfo.setIccardno(medicalInfoData.getString("cardno"));
		medicalInfo.setIndex(medicalInfoData.getInt("index"));
		medicalInfo.setType(medicalInfoData.getInt("type"));
		SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
		Date date = format.parse(medicalInfoData.getString("datetime"));
		medicalInfo.setDateTime(date);
		medicalInfo.setValue(Double.valueOf(medicalInfoData.getInt("value")));
		medicalInfo.setRemark(medicalInfoData.getString("remark"));
		medicalInfo.setShopName(medicalInfoData.getString("shopname"));
		medicalInfo.setShopCode(medicalInfoData.getString("shopcode"));
		return medicalInfo;
	}
}